package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dwd.db.app
 * @className: DwdTradeOrderRefund
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/27 9:30
 * @version: 1.0
 */

public class DwdTradeOrderRefund extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderRefund().start(Constant.TOPIC_DB,Constant.TOPIC_DWD_TRADE_ORDER_REFUND,4,10017);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //设置ttl
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //读取topic_db
        readOdsDb(tableEnv,Constant.TOPIC_DWD_TRADE_ORDER_REFUND);
        //读取dic
        readBaseDic(tableEnv);
        // 2. 过滤退单表数据 order_refund_info   insert
        Table orderRefundInfo = getOrderRefundInfo(tableEnv);
        tableEnv.createTemporaryView("order_refund_info",orderRefundInfo);
//        orderRefundInfo.execute().print();
        // 3. 过滤订单表中的退单数据: order_info  update
        Table getOrderInfo = getGetOrderInfo(tableEnv);
        tableEnv.createTemporaryView("order_info",getOrderInfo);
//        getOrderInfo.execute().print();
        // 4. join: 普通的和 lookup join
        Table joinTable = getJoinTable(tableEnv);
//        joinTable.execute().print();

        // 写入kfk
        extracted(tableEnv);
        joinTable.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_REFUND);


    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table "+ Constant.TOPIC_DWD_TRADE_ORDER_REFUND+"(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    order_id string,\n" +
                "    sku_id string,\n" +
                "    province_id string,\n" +
                "    refund_type_code string,\n" +
                "    refund_type_name string,\n" +
                "    refund_num string,\n" +
                "    refund_amount string,\n" +
                "    refund_reason_type_code string,\n" +
                "    refund_reason_type_name string,\n" +
                "    refund_reason_txt string,\n" +
                "    refund_status string,\n" +
                "    order_status string,\n" +
                "    date_id string,\n" +
                "    create_time string,\n" +
                "    ts bigint\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
    }

    private static Table getJoinTable(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select ori.id,\n" +
                "       ori.user_id,\n" +
                "       ori.order_id,\n" +
                "       ori.sku_id,\n" +
                "       oi.province_id,\n" +
                "       ori.refund_type as refund_type_code,\n" +
                "       c1.info.dic_name as refund_type_name,\n" +
                "       ori.refund_num,\n" +
                "       ori.refund_amount,\n" +
                "       ori.refund_reason_type as refund_reason_type_code,\n" +
                "       c2.info.dic_name as refund_reason_type_name,\n" +
                "       ori.refund_reason_txt,\n" +
                "       ori.refund_status,\n" +
                "       oi.order_status,\n" +
                "       date_format(ori.create_time,'yyyy-MM-dd') date_id,\n" +
                "       ori.create_time,\n" +
                "       ts\n" +
                "from order_refund_info ori\n" +
                "join order_info oi on oi.id = ori.order_id\n" +
                "join base_dic  FOR SYSTEM_TIME AS OF ori.proc_time AS c1 ON ori.refund_type = c1.rowkey\n" +
                "join base_dic  FOR SYSTEM_TIME AS OF ori.proc_time AS c2 ON ori.refund_reason_type = c2.rowkey");
    }

    private static Table getGetOrderInfo(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['order_status'] order_status,\n" +
                "    `data`['province_id'] province_id\n" +
                "from topic_db where `database` = 'gmall'\n" +
                "and `table` = 'order_info'\n" +
                "and `type` = 'update'\n" +
                "and `old`['order_status'] is not null\n" +
                "and `data`['order_status'] = '1005'");
    }

    private static Table getOrderRefundInfo(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select `data`['id']            id,\n" +
                "       `data`['user_id']       user_id,\n" +
                "       `data`['order_id']      order_id,\n" +
                "       `data`['sku_id']        sku_id,\n" +
                "       `data`['refund_type']   refund_type,\n" +
                "       `data`['refund_num']    refund_num,\n" +
                "       `data`['refund_amount'] refund_amount,\n" +
                "       `data`['refund_reason_type'] refund_reason_type,\n" +
                "       `data`['refund_reason_txt'] refund_reason_txt,\n" +
                "       `data`['refund_status'] refund_status,\n" +
                "       `data`['create_time'] create_time,\n" +
                "       proc_time,\n" +
                "       ts\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "  and `table` = 'order_refund_info'\n" +
                "  and `type` = 'insert';");
    }
}
